package com.pickup.kafka.controller;

import com.pickup.kafka.consumer.KafkaConsumer;
import com.pickup.kafka.provider.KafkaProvider;
import com.pickup.pool.ThreadPoolUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;


@RestController
public class kafkaController {


    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private KafkaListenerEndpointRegistry registry;

    @Autowired
    private KafkaConsumer kafkaConsumer;

    @Autowired
    private ThreadPoolUtil threadPoolUtil;

//    @Autowired
//    private AdminClient adminClient;

    @Autowired
    private KafkaProvider kafkaProvider;

    @GetMapping("/sendKafka")
    public void sendKafka(String message) throws Exception {
        registry.getListenerContainer("task").stop();
        threadPoolUtil.getExecutorPool().execute(() -> {
            try {
                for (int i = 0; i < 20; i++) {
                    //kafkaProvider.sendChatMsg("ChatMsg", "老许", message);
                   // kafkaProvider.sendChatMsg("ChatMsg", "老王", message);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

//    @GetMapping("/receiveKafka1")
//    public void receiveKafka1() throws Exception {
//        registry.getListenerContainer("task").start();
//        kafkaConsumer.startConsumer("老许", "task");
//    }
//
//    @GetMapping("/receiveKafka2")
//    public void receiveKafka2() throws Exception {
//        registry.getListenerContainer("task").start();
//        kafkaConsumer.startConsumer("老王", "task");
//    }

}
